package com.alibaba.cirtus.ots.application.adapter.ability.facade.serviceV2.kafka;

import com.alibaba.cz.base.tool.log.Log;
import lombok.Data;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Component;


import java.util.HashMap;
import java.util.Map;

/**
 * yufeng
 * 2023/6/1
 **/
@Data
@Component
@Configuration
public class KafkaProducerProperties {
    private static final Log log = Log.getLogger(KafkaProducerProperties.class);
    private String servers="10.240.4.35:9092,10.240.4.37:9092,10.240.4.36:9092";
    private int retries=0;
    private int batchSize=4096;
    private int linger=1;
    private int bufferMemory=4096000;
    private int maxRequestSizeConfig=3000000;
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
//        servers="10.240.4.35:9092,10.240.4.37:9092,10.240.4.36:9092";

        servers ="alikafka-post-cn-uax36uo7c001-1-vpc.alikafka.aliyuncs.com:9092," +
                "alikafka-post-cn-uax36uo7c001-2-vpc.alikafka.aliyuncs.com:9092," +
                "alikafka-post-cn-uax36uo7c001-3-vpc.alikafka.aliyuncs.com:9092";
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSizeConfig);
        //props.put(ProducerConfig.ACKS_CONFIG, 1);
        return props;
    }
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate kafkaTemplate = null;
        try{
            kafkaTemplate = new KafkaTemplate(producerFactory());
            log.info("KafkaTemplate:{}",kafkaTemplate);
            return  kafkaTemplate;
        }catch (Exception e){
            return null;
        }

    }

}
